package io.github.libkodi.session;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.github.libkodi.basic_gateway.component.AbstractCompoent;
import io.github.libkodi.session.interfaces.Session;
import io.vertx.core.Future;
import io.vertx.core.json.JsonObject;
import io.vertx.redis.client.Command;
import io.vertx.redis.client.Redis;
import io.vertx.redis.client.RedisAPI;
import io.vertx.redis.client.Request;

public class DataCenter extends AbstractCompoent {
	private static final Logger log = LoggerFactory.getLogger(DataCenter.class);
	
	private HashMap<String, DataItem> map = new HashMap<String, DataItem>();
	private Redis redis = null;
	private String suffix = "";
	private long id = -1;
	
	@Override
	public void init() {
		Object period = ctx.configValue("server.session.period");
		
		period = String.valueOf(period);
		
		if (period == null || !Pattern.matches("^\\d+$", (String)period)) {
			period = "3600";
		}
		
		if (this.id > 0) {
			ctx.vertx().cancelTimer(this.id);
		}
		
		this.id = ctx.vertx().setPeriodic(Integer.parseInt((String) period) * 1000, id -> {
			this.id = id;
			this.reclaimMemory();
		});
	}
	
	private void reclaimMemory() {
		try {
			ArrayList<String> keys = new ArrayList<String>();
			Iterator<Entry<String, DataItem>> iter = map.entrySet().iterator();
			
			while (iter.hasNext()) {
				Entry<String, DataItem> ent = iter.next();
				
				if (ent.getValue().isExpired()) {
					keys.add(ent.getKey());
				}
			}
			
			synchronized (map) {
				for (String key : keys) {
					try {
						map.remove(key);
					} catch (Exception e) {
					}
				}
			}
		} catch (Exception e) {
			log.error("Error reclaiming memory:", e);
		}
	}
	
	public DataCenter addSuffix(String suffix) {
		this.suffix += suffix + ":";
		return this;
	}
	
	public DataCenter useRedis(Redis redis) {
		return useRedis(redis, false);
	}
	
	public DataCenter useRedis(Redis redis, boolean autoClose) {
		if (this.redis != null && autoClose) {
			this.redis.close();
		}
		
		this.redis = redis;
		
		if (ctx != null && id > 0) {
			ctx.vertx().cancelTimer(id);
			id = -1;
		}
		
		return this;
	}
	
	private Future<String> formatValue(Object value) {
		return Future.future(promise -> {
			Object v = value;
			
			try {
				if (v instanceof byte[]) {
					v = new String((byte[]) v);
				} else if (v instanceof Boolean) {
					if (v.equals(true)) {
						v = "t";
					} else {
						v = "f";
					}
				}
				
				if (!(value instanceof String)) {
					v = String.valueOf(v);
				}
				
				promise.complete((String) v);
			} catch (Exception e) {
				promise.fail(e);
			}
		});
	}

	private byte[] objectToBytes(Object value) throws Exception {
		ByteArrayOutputStream bos = new ByteArrayOutputStream();
		ObjectOutputStream oos = null;
		byte[] res = null;
		Exception error = null;
		
		try {
			oos = new ObjectOutputStream(bos);
			oos.writeObject(value);
			res = bos.toByteArray();
		} catch (Exception e) {
			error = e;
		} finally {
			try {
				bos.close();
			} catch (IOException e) {
			}
			
			if (oos != null) {
				try {
					oos.close();
				} catch (IOException e) {
				}
			}
		}
		
		if (error != null) {
			throw error;
		} else {
			return res;
		}
	}
	
	private Future<Void> setValue(String key, Object value) {
		return Future.future(promise -> {
			if (redis != null) {
				formatValue(value).onSuccess(v -> {
					ArrayList<String> args = new ArrayList<String>();
					args.add(this.suffix + key);
					args.add(v);
					
					RedisAPI.api(redis).set(args).onSuccess(resp -> {
						promise.complete();
					}).onFailure(promise::fail);
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					try {
						String k = this.suffix + key;
						
						if (!map.containsKey(k)) {
							map.put(k, new DataItem(value));
						} else {
							DataItem item = map.get(k);
							item.setObject(value);
						}
						
						promise.complete();
					} catch (Exception e) {
						promise.fail(e);
					}
				}
			}
		});
	}
	
	public Future<Void> setString(String key, String value) {
		return setValue(key, value);
	}
	
	public Future<Void> setInteger(String key, int value) {
		return setValue(key, value);
	}
	
	public Future<Void> setDouble(String key, double value) {
		return setValue(key, value);
	}
	
	public Future<Void> setFloat(String key, float value) {
		return setValue(key, value);
	}
	
	public Future<Void> setLong(String key, long value) {
		return setValue(key, value);
	}

	public Future<Void> setBoolean(String key, boolean value) {
		return setValue(key, value);
	}
	
	public Future<Void> setJavaObject(String key, Object value) {
		return Future.future(promise -> {
			if (!Serializable.class.isAssignableFrom(value.getClass())) {
				promise.fail("Object must implement Serializable interface.");
			} else {
				if (redis != null) {
					try {
						byte[] bytes = objectToBytes(value);
						
						redis.connect().onSuccess(conn -> {
							conn.send(Request.cmd(Command.SET, this.suffix + key, bytes)).onSuccess(resp -> {
								conn.close();
								promise.complete();
							}).onFailure(e -> {
								conn.close();
								promise.fail(e);
							});
						}).onFailure(promise::fail);
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					setValue(key, value).onSuccess(v -> {
						promise.complete();
					}).onFailure(promise::fail);
				}
			}
		});
	}
	
	private Future<Void> setHashValue(String key, String field, Object value) {
		return Future.future(promise -> {
			if (redis != null) {
				formatValue(value).onSuccess(v -> {
					ArrayList<String> args = new ArrayList<String>();
					args.add(this.suffix + key);
					args.add(field);
					args.add(v);
					
					RedisAPI.api(redis).hset(args).onSuccess(resp -> {
						promise.complete();
					}).onFailure(promise::fail);
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					try {
						String k = this.suffix + key;
						
						if (!map.containsKey(k)) {
							map.put(k, new DataItem(new JsonObject().put(field, value)));
						} else {
							DataItem item = map.get(k);
							item.getMap().put(field, value);
						}
						
						promise.complete();
					} catch (Exception e) {
						promise.fail(e);
					}
				}
			}
		});
	}
	
	public Future<Void> setHashString(String key, String field, String value) {
		return setHashValue(key, field, value);
	}
	
	public Future<Void> setHashInteger(String key, String field, int value) {
		return setHashValue(key, field, value);
	}
	
	public Future<Void> setHashDouble(String key, String field, double value) {
		return setHashValue(key, field, value);
	}
	
	public Future<Void> setHashFloat(String key, String field, float value) {
		return setHashValue(key, field, value);
	}
	
	public Future<Void> setHashLong(String key, String field, long value) {
		return setHashValue(key, field, value);
	}
	
	public Future<Void> setHashBoolean(String key, String field, boolean value) {
		return setHashValue(key, field, value);
	}
	
	public Future<Void> setHashJavaObject(String key, String field, Object value) {
		return Future.future(promise -> {
			if (!Serializable.class.isAssignableFrom(value.getClass())) {
				promise.fail("Object must implement Serializable interface.");
			} else {
				if (redis != null) {
					try {
						byte[] bytes = objectToBytes(value);
						
						redis.connect().onSuccess(conn -> {
							conn.send(Request.cmd(Command.HSET, this.suffix + key, field, bytes)).onSuccess(resp -> {
								conn.close();
								promise.complete();
							}).onFailure(e -> {
								conn.close();
								promise.fail(e);
							});
						}).onFailure(promise::fail);
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					setHashValue(key, field, value).onSuccess(v -> {
						promise.complete();
					}).onFailure(promise::fail);
				}
			}
		});
	}
	
	public Future<Void> setExpire(String key, int seconds) {
		return Future.future(promise -> {
			if (redis != null) {
				ArrayList<String> args = new ArrayList<String>();
				args.add(this.suffix + key);
				args.add(String.valueOf(seconds));
				
				RedisAPI.api(redis).expire(args).onSuccess(resp -> {
					promise.complete();
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					try {
						String k = this.suffix + key;
						
						if (map.containsKey(k)) {
							map.get(k).setExpire(seconds);
						}
						
						promise.complete();
					} catch (Exception e) {
						promise.fail(e);
					}
				}
			}
		});
	}
	
	private Future<Object> getValue(String key) {
		return getValue(key, false);
	}
	
	private Future<Object> getValue(String key, boolean raw) {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI.api(redis).get(this.suffix + key).onSuccess(resp -> {
					if (resp != null) {
						promise.complete(raw ? resp.toBytes() : resp.toString());
					} else {
						promise.complete(null);
					}
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					try {
						String k = this.suffix + key;
						
						if (map.containsKey(k)) {
							DataItem item = map.get(k);
							
							if (!item.isExpired()) {
								promise.complete(item.getObject());
							} else {
								try {
									map.remove(k);
								} catch (Exception e) {
								}
							}
						}
						
						promise.complete(null);
					} catch (Exception e) {
						promise.fail(e);
					}
				}
			}
		});
	}
	
	public Future<String> getString(String key) {
		return Future.future(promise -> {
			getValue(key).onSuccess(val -> {
				if (redis != null) {
					promise.complete((String) val);
				} else {
					promise.complete((String) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Integer> getInteger(String key) {
		return Future.future(promise -> {
			getValue(key).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Integer.parseInt((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((int) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Double> getDouble(String key) {
		return Future.future(promise -> {
			getValue(key).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Double.parseDouble((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((double) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Float> getFloat(String key) {
		return Future.future(promise -> {
			getValue(key).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Float.parseFloat((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((float) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Long> getLong(String key) {
		return Future.future(promise -> {
			getValue(key).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Long.parseLong((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((long) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Boolean> getBoolean(String key) {
		return Future.future(promise -> {
			getValue(key).onSuccess(val -> {
				if (redis != null) {
					try {
						if ("t".equals(val)) {
							promise.complete(true);
						} else {
							promise.complete(false);
						}
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((boolean) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	@SuppressWarnings("unchecked")
	public <T> Future<T> getJavaObject(String key, Class<T> cls) {
		return Future.future(promise -> {
			getValue(key, true).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(parseJavaObject((byte[]) val, cls));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((T) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	@SuppressWarnings("unchecked")
	private <T> T parseJavaObject(byte[] val, Class<T> cls) throws Exception {
		ByteArrayInputStream bis = null;
		Exception error = null;
		T res = null;
		ObjectInputStream ois = null;
		
		try {
			bis = new ByteArrayInputStream(val);
			ois = new ObjectInputStream(bis);
			res = (T) ois.readObject();
		} catch (Exception e) {
			error = e;
		} finally {
			if (bis != null) {
				try {
					bis.close();
				} catch (IOException e) {
				}
			}
			
			if (ois != null) {
				try {
					ois.close();
				} catch (IOException e) {
				}
			}
		}
		
		if (error != null) {
			throw error;
		} else {
			return res;
		}
	}
	
	private Future<Object> getHashValue(String key, String filed) {
		return getHashValue(key, filed, false);
	}
	
	private Future<Object> getHashValue(String key, String filed, boolean raw) {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI.api(redis).hget(this.suffix + key, filed).onSuccess(resp -> {
					if (resp != null) {
						promise.complete(raw ? resp.toBytes() : resp.toString());
					} else {
						promise.complete(null);
					}
				}).onFailure(promise::fail);
			} else {
				getValue(key).onSuccess(obj -> {
					if (obj != null) {
						try {
							promise.complete(((JsonObject) obj).getValue(filed));
						} catch (Exception e) {
							promise.fail(e);
						}
					} else {
						promise.complete(null);
					}
				}).onFailure(promise::fail);
			}
		});
	}
	
	public Future<String> getHashString(String key, String field) {
		return Future.future(promise -> {
			getHashValue(key, field).onSuccess(val -> {
				promise.complete((String) val);
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Integer> getHashInteger(String key, String field) {
		return Future.future(promise -> {
			getHashValue(key, field).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Integer.parseInt((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((int) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Double> getHashDouble(String key, String field) {
		return Future.future(promise -> {
			getHashValue(key, field).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Double.parseDouble((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((double) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Float> getHashFloat(String key, String field) {
		return Future.future(promise -> {
			getHashValue(key, field).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Float.parseFloat((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((float) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Long> getHashLong(String key, String field) {
		return Future.future(promise -> {
			getHashValue(key, field).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(Long.parseLong((String) val));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((long) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	public Future<Boolean> getHashBoolean(String key, String field) {
		return Future.future(promise -> {
			getHashValue(key, field).onSuccess(val -> {
				if (redis != null) {
					try {
						if ("t".equals(val)) {
							promise.complete(true);
						} else {
							promise.complete(false);
						}
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((boolean) val);
				}
			}).onFailure(promise::fail);
		});
	}
	
	@SuppressWarnings("unchecked")
	public <T> Future<T> getHashJavaObject(String key, String field, Class<T> cls) {
		return Future.future(promise -> {
			getHashValue(key, field, true).onSuccess(val -> {
				if (redis != null) {
					try {
						promise.complete(parseJavaObject((byte[]) val, cls));
					} catch (Exception e) {
						promise.fail(e);
					}
				} else {
					promise.complete((T) val);
				}
			}).onFailure(promise::fail);
		});
	}

	public Future<Set<String>> keys() {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI.api(redis).keys("*").onSuccess(resp -> {
					ArrayList<String> list = new ArrayList<String>();
					
					for (int i = 0, len = resp.size(); i < len; i ++) {
						list.add(resp.get(i).toString());
					}
					
					promise.complete(Set.copyOf(list));
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					promise.complete(map.keySet());
				}
			}
		});
	}
	
	public Future<Set<String>> keys(String key) {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI.api(redis).hkeys(this.suffix + key).onSuccess(resp -> {
					ArrayList<String> list = new ArrayList<String>();
					
					for (int i = 0, len = resp.size(); i < len; i ++) {
						list.add(resp.get(i).toString());
					}
					
					promise.complete(Set.copyOf(list));
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					promise.complete(map.get(this.suffix + key).getMap().keySet());
				}
			}
		});
	}
	
	public Future<Void> remove(String key) {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI.api(redis).del(List.of(this.suffix + key)).onSuccess(resp -> {
					promise.complete();
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					try {
						map.remove(this.suffix + key);
						promise.complete();
					} catch (Exception e) {
						promise.fail(e);
					}
				}
			}
		});
	}
	
	public Future<Void> remove(String key, String field) {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI.api(redis).hdel(List.of(this.suffix + key, field)).onSuccess(resp -> {
					promise.complete();
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					try {
						DataItem item = map.get(this.suffix + key);
						
						if (item != null) {
							item.getMap().remove(field);
							promise.complete();
						} else {
							promise.complete();
						}
					} catch (Exception e) {
						promise.fail(e);
					}
				}
			}
		});
	}
	
	public Future<Session> session(String id) {
		return session(id, false);
	}
	
	public Future<Session> session(String id, boolean autocreate) {
		return Future.future(promise -> {
			if (redis != null) {
				RedisAPI api = RedisAPI.api(redis);
				
				api.exists(List.of(this.suffix + id)).onSuccess(resp -> {
					try {
						if (resp.toInteger() == 0) {
							if (!autocreate) {
								promise.complete(null);
							}
						}
						
						if (!promise.future().isComplete()) {
							promise.complete(new SessionImpl(id, this));
						}
					} catch (Exception e) {
						promise.fail(e);
					}
				}).onFailure(promise::fail);
			} else {
				synchronized (map) {
					String k = this.suffix + id;
					
					if (!map.containsKey(k)) {
						if (!autocreate) {
							promise.complete(null);
						}
					}
					
					if (!promise.future().isComplete()) {
						promise.complete(new SessionImpl(id, this));
					}
				}
			}
		});
	}
}
